// Copyright(c) 2022 to 2023 ZettaScale Technology and others
//
// This program and the accompanying materials are made available under the
// terms of the Eclipse Public License v. 2.0 which is available at
// http://www.eclipse.org/legal/epl-2.0, or the Eclipse Distribution License
// v. 1.0 which is available at
// http://www.eclipse.org/org/documents/edl-v10.php.
//
// SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <assert.h>
#include <locale.h>
#include <signal.h>

#include "dds/dds.h"
#include "dynsub.h"

// Interpreting the data of an arbitrary topic requires interpreting the type object that describes the data.
// The type object type is defined by the XTypes specification (https://www.omg.org/spec/DDS-XTypes/) and it
// comes in two forms: MinimalTypeObject and CompleteTypeObject.  Only the latter includes field names, so
// that's what need.
//
// Cyclone DDS includes a copy of the IDL as well as the corresponding type definitions in C as generated by
// IDLC.  So instead of including yet another copy, we simply refer to those.  These files are not (yet?)
// part of the stable API of Cyclone DDS and so the updates to Cyclone may change the locations or the names
// of the relevant header files.
//
// The API uses `dds_typeobj_t` and `dds_typeinfo_t` that are opaque types but really amount to the
// corresponding XTypes objects: DDS_XTypes_TypeObject and DDS_XTypes_TypeInformation.  Rather than casting
// pointers like we do here, they should be defined in a slightly different way so that they are not really
// opaque.  For now, this'll have to do.
#include "dds/ddsc/dds_public_alloc.h"
#include "dds/ddsi/ddsi_sertype.h"
#include "dds/ddsi/ddsi_xt_typeinfo.h"

#include "dds/ddsrt/threads.h"
#include "dds/ddsi/ddsi_serdata.h"

// For convenience, the DDS participant is global
static dds_entity_t participant;

static dds_entity_t termcond;

// Helper function to wait for a DCPSPublication/DCPSSubscription to show up with the desired topic name,
// then calls dds_find_topic to create a topic for that data writer's/reader's type up the retrieves the
// type object.
static dds_return_t get_topic_and_typeobj (const char *topic_name, dds_duration_t timeout, dds_entity_t *topic, DDS_XTypes_TypeObject **xtypeobj)
{
  const dds_entity_t waitset = dds_create_waitset (participant);
  const dds_entity_t dcpspublication_reader = dds_create_reader (participant, DDS_BUILTIN_TOPIC_DCPSPUBLICATION, NULL, NULL);
  const dds_entity_t dcpspublication_readcond = dds_create_readcondition (dcpspublication_reader, DDS_ANY_STATE);
  const dds_entity_t dcpssubscription_reader = dds_create_reader (participant, DDS_BUILTIN_TOPIC_DCPSSUBSCRIPTION, NULL, NULL);
  const dds_entity_t dcpssubscription_readcond = dds_create_readcondition (dcpssubscription_reader, DDS_ANY_STATE);
  (void) dds_waitset_attach (waitset, dcpspublication_readcond, dcpspublication_reader);
  (void) dds_waitset_attach (waitset, dcpssubscription_readcond, dcpssubscription_reader);
  const dds_time_t abstimeout = (timeout == DDS_INFINITY) ? DDS_NEVER : dds_time () + timeout;
  dds_return_t ret = DDS_RETCODE_OK;
  *xtypeobj = NULL;
  struct ppc ppc;
  ppc_init (&ppc);
  dds_attach_t triggered_reader_x;
  while (*xtypeobj == NULL && dds_waitset_wait_until (waitset, &triggered_reader_x, 1, abstimeout) > 0)
  {
    void *epraw = NULL;
    dds_sample_info_t si;
    dds_entity_t triggered_reader = (dds_entity_t) triggered_reader_x;
    if (dds_take (triggered_reader, &epraw, &si, 1, 1) <= 0)
      continue;
    dds_builtintopic_endpoint_t *ep = epraw;
    const dds_typeinfo_t *typeinfo = NULL;
    // We are only interested in DCPSPublications where the topic name matches and that carry type information
    // (a non-XTypes capable DDS would not provide type information) because without that information there is
    // no way we can do anything interesting with it.
    if (strcmp (ep->topic_name, topic_name) == 0 && dds_builtintopic_get_endpoint_type_info (ep, &typeinfo) == 0 && typeinfo)
    {
      // Using dds_find_topic allows us to "clone" the topic definition including the topic QoS, but it does
      // require that topic discovery is enabled in the configuration.  The advantage of using dds_find_topic
      // is that it creates a topic with the same name, type *and QoS*.  That distinction only matters if
      // topic is discovery is enabled and/or if the topic has a durability kind of of transient or persistent:
      // - using a different topic QoS might result in an incompatible QoS notification if topic discovery is
      //   enabled (everything would still work).
      // - transient/persistent data behaviour is defined in terms of the topic QoS actually really matters
      //
      // So we try to use dds_find_topic, and if that fails, try to go the other route using the writer's QoS
      // as an approximation of the topic QoS.
      if ((*topic = dds_find_topic (DDS_FIND_SCOPE_GLOBAL, participant, ep->topic_name, typeinfo, DDS_SECS (2))) < 0)
      {
        fprintf (stderr, "dds_find_topic: %s ... continuing on the assumption that topic discovery is disabled\n", dds_strretcode (*topic));
        dds_topic_descriptor_t *descriptor;
        if ((ret = dds_create_topic_descriptor(DDS_FIND_SCOPE_GLOBAL, participant, typeinfo, DDS_SECS (10), &descriptor)) < 0)
        {
          fprintf (stderr, "dds_create_topic_descriptor: %s\n", dds_strretcode (ret));
          dds_return_loan (triggered_reader, &epraw, 1);
          goto error;
        }
        dds_qset_data_representation (ep->qos, 0, NULL);
        if ((*topic = dds_create_topic (participant, descriptor, ep->topic_name, ep->qos, NULL)) < 0)
        {
          fprintf (stderr, "dds_create_topic_descriptor: %s (be sure to enable topic discovery in the configuration)\n", dds_strretcode (*topic));
          dds_delete_topic_descriptor (descriptor);
          dds_return_loan (triggered_reader, &epraw, 1);
          goto error;
        }
        dds_delete_topic_descriptor (descriptor);
      }
      // The topic suffices for creating a reader, but we also need the TypeObject to make sense of the data
      if ((*xtypeobj = load_type_with_deps (participant, typeinfo, &ppc)) == NULL)
      {
        fprintf (stderr, "loading type with all dependencies failed\n");
        dds_return_loan (triggered_reader, &epraw, 1);
        goto error;
      }
      if (load_type_with_deps_min (participant, typeinfo, &ppc) == NULL)
      {
        fprintf (stderr, "loading minimal type with all dependencies failed\n");
        dds_return_loan (triggered_reader, &epraw, 1);
        goto error;
      }
    }
    dds_return_loan (triggered_reader, &epraw, 1);
  }
  if (*xtypeobj)
  {
    // If we got the type object, populate the type cache
    size_t align, size;
    build_typecache_to (&(*xtypeobj)->_u.complete, &align, &size);
    fflush (stdout);
    struct typeinfo templ = { .key = { .key = (uintptr_t) *xtypeobj } } , *info;
    if ((info = type_cache_lookup (&templ)) != NULL)
    {
      assert (info->release == NULL);
      info->release = *xtypeobj;
    }
    else
    {
      // not sure whether this is at all possible
      info = malloc (sizeof (*info));
      assert (info);
      *info = (struct typeinfo){ .key = { .key = (uintptr_t) *xtypeobj }, .typeobj = &(*xtypeobj)->_u.complete, .release = *xtypeobj, .align = align, .size = size };
      type_cache_add (info);
    }
  }
error:
  dds_delete (dcpspublication_reader);
  dds_delete (dcpssubscription_reader);
  dds_delete (waitset);
  return (*xtypeobj != NULL) ? DDS_RETCODE_OK : DDS_RETCODE_TIMEOUT;
}

static bool print_sample_normal (dds_entity_t reader, const DDS_XTypes_TypeObject *xtypeobj)
{
  void *raw = NULL;
  dds_sample_info_t si;
  dds_return_t ret;
  if ((ret = dds_take (reader, &raw, &si, 1, 1)) < 0)
    return false;
  else if (ret != 0)
  {
    // ... that we then print
    print_sample (si.valid_data, raw, &xtypeobj->_u.complete);
    if (dds_return_loan (reader, &raw, 1) < 0)
      return false;
  }
  return true;
}

static void hexdump (const unsigned char *msg, const size_t len)
{
  for (size_t off16 = 0; off16 < len; off16 += 16)
  {
    printf ("%04" PRIxSIZE " ", off16);
    size_t off1;
    for (off1 = 0; off1 < 16 && off16 + off1 < len; off1++)
      printf ("%s %02x", (off1 == 8) ? " " : "", msg[off16 + off1]);
    for (; off1 < 16; off1++)
      printf ("%s   ", (off1 == 8) ? " " : "");
    printf ("  |");
    for (off1 = 0; off1 < 16 && off16 + off1 < len; off1++)
    {
      unsigned char c = msg[off16 + off1];
      printf ("%c", (c >= 32 && c < 127) ? c : '.');
    }
    printf ("|\n");
  }
  fflush (stdout);
}

static const char *encodingstr (const struct ddsi_serdata *sd)
{
  uint16_t encoding;
  ddsi_serdata_to_ser (sd, 0, 2, &encoding);
  switch (encoding)
  {
    case DDSI_RTPS_CDR_BE:
    case DDSI_RTPS_CDR_LE:
      return "CDR";
    case DDSI_RTPS_PL_CDR_BE:
    case DDSI_RTPS_PL_CDR_LE:
      return "PL_CDR";
    case DDSI_RTPS_CDR2_BE:
    case DDSI_RTPS_CDR2_LE:
      return "CDR2";
    case DDSI_RTPS_D_CDR2_BE:
    case DDSI_RTPS_D_CDR2_LE:
      return "D_CDR2";
    case DDSI_RTPS_PL_CDR2_BE:
    case DDSI_RTPS_PL_CDR2_LE:
      return "PL_CDR2";
    default:
      return "unknown";
  }
}

static bool print_sample_cdr (dds_entity_t reader, const DDS_XTypes_TypeObject *xtypeobj)
{
  // Note: doesn't print the exact CDR received, but the "normalised" one where byteswapping
  // has been performed, booleans have been mapped to 0 or 1, and perhaps some other similar
  // changes have been made.
  struct ddsi_serdata *sd = NULL;
  dds_sample_info_t si;
  dds_return_t ret;
  if ((ret = dds_takecdr (reader, &sd, 1, &si, 0)) < 0)
    return false;
  else if (ret != 0)
  {
    printf ("encoding: %s", encodingstr (sd));
    if (!si.valid_data)
      printf (" (expect XCDR2 because it is an invalid sample)");
    printf ("\n");
    if (ddsi_serdata_size (sd) == 4)
      printf ("(no payload)\n");
    else
    {
      ddsrt_iovec_t iov;
      struct ddsi_serdata *refsd;
      refsd = ddsi_serdata_to_ser_ref (sd, 4, ddsi_serdata_size (sd) - 4, &iov);
      hexdump (iov.iov_base, iov.iov_len);
      ddsi_serdata_to_ser_unref (refsd, &iov);
    }

    void *raw = calloc (1, sd->type->sizeof_type);
    if (raw == NULL)
      abort ();

    bool ok;
    if (si.valid_data)
      ok = ddsi_serdata_to_sample (sd, raw, NULL, NULL);
    else
    {
      const struct ddsi_sertype *st;
      dds_get_entity_sertype (reader, &st);
      ok = ddsi_serdata_untyped_to_sample (st, sd, raw, NULL, NULL);
    }
    if (ok)
      print_sample (si.valid_data, raw, &xtypeobj->_u.complete);
    else
      printf ("(conversion to sample failed)\n");
    ddsi_sertype_free_sample (sd->type, raw, DDS_FREE_CONTENTS);
    free (raw);

    ddsi_serdata_unref (sd);
  }
  return true;
}

#if !DDSRT_WITH_FREERTOS && !__ZEPHYR__
static void signal_handler (int sig)
{
  (void) sig;
  dds_set_guardcondition (termcond, true);
}
#endif

#if !_WIN32 && !DDSRT_WITH_FREERTOS && !__ZEPHYR__
static uint32_t sigthread (void *varg)
{
  sigset_t *set = varg;
  int sig;
  if (sigwait (set, &sig) == 0)
    signal_handler (sig);
  return 0;
}
#endif

int main (int argc, char **argv)
{
  dds_return_t ret = 0;
  dds_entity_t topic = 0;
  bool raw_mode;
  const char *topic_name;

  // for printf("%ls")
  setlocale (LC_CTYPE, "");

  if (argc == 2)
  {
    raw_mode = false;
    topic_name = argv[1];
  }
  else if (argc == 3 && strcmp (argv[1], "-r") == 0)
  {
    raw_mode = true;
    topic_name = argv[2];
  }
  else
  {
    fprintf (stderr, "usage: %s [-r] topicname\n", argv[0]);
    return 2;
  }

  participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);
  if (participant < 0)
  {
    fprintf (stderr, "dds_create_participant: %s\n", dds_strretcode (participant));
    return 1;
  }

  // The one magic step: get a topic and type object ...
  DDS_XTypes_TypeObject *xtypeobj;
  type_cache_init ();
  if ((ret = get_topic_and_typeobj (topic_name, DDS_SECS (10), &topic, &xtypeobj)) < 0)
  {
    fprintf (stderr, "get_topic_and_typeobj: %s\n", dds_strretcode (ret));
    goto error;
  }
  // ... given those, we can create a reader just like we do normally ...
  const dds_entity_t reader = dds_create_reader (participant, topic, NULL, NULL);
  // ... and create a waitset that allows us to wait for any incoming data ...
  const dds_entity_t waitset = dds_create_waitset (participant);
  const dds_entity_t readcond = dds_create_readcondition (reader, DDS_ANY_STATE);
  (void) dds_waitset_attach (waitset, readcond, 0);

  termcond = dds_create_guardcondition (participant);
  (void) dds_waitset_attach (waitset, termcond, 0);

#ifdef _WIN32
  signal (SIGINT, signal_handler);
#elif !DDSRT_WITH_FREERTOS && !__ZEPHYR__
  ddsrt_thread_t sigtid;
  sigset_t sigset, osigset;
  sigemptyset (&sigset);
#ifdef __APPLE__
  DDSRT_WARNING_GNUC_OFF(sign-conversion)
#endif
  sigaddset (&sigset, SIGHUP);
  sigaddset (&sigset, SIGINT);
  sigaddset (&sigset, SIGTERM);
#ifdef __APPLE__
  DDSRT_WARNING_GNUC_ON(sign-conversion)
#endif
  sigprocmask (SIG_BLOCK, &sigset, &osigset);
  {
    ddsrt_threadattr_t tattr;
    ddsrt_threadattr_init (&tattr);
    ddsrt_thread_create (&sigtid, "sigthread", &tattr, sigthread, &sigset);
  }
#endif

  bool termflag = false;
  while (!termflag)
  {
    (void) dds_waitset_wait (waitset, NULL, 0, DDS_INFINITY);
    dds_read_guardcondition (termcond, &termflag);

    bool ok = raw_mode ? print_sample_cdr (reader, xtypeobj) : print_sample_normal (reader, xtypeobj);
    if (!ok)
      break;
  }

#if _WIN32
  signal_handler (SIGINT);
#elif !DDSRT_WITH_FREERTOS && !__ZEPHYR__
  {
    /* get the attention of the signal handler thread */
    void (*osigint) (int);
    void (*osigterm) (int);
    kill (getpid (), SIGTERM);
    ddsrt_thread_join (sigtid, NULL);
    osigint = signal (SIGINT, SIG_IGN);
    osigterm = signal (SIGTERM, SIG_IGN);
    sigprocmask (SIG_SETMASK, &osigset, NULL);
    signal (SIGINT, osigint);
    signal (SIGINT, osigterm);
  }
#endif

error:
  type_cache_free ();
  dds_delete (participant);
  return ret < 0;
}
